#include <klippy/reactor.h>
#include <sys/epoll.h>
#include <unistd.h>
#include <vector>
#include <map>
#include <mutex>
#include <queue>
#include <system_error>
#include <iostream>

namespace klippy {

class EPollReactor : public Reactor {
public:
    EPollReactor() : running_(false), epoll_fd_(-1) {
        epoll_fd_ = epoll_create1(0);
        if (epoll_fd_ == -1) {
            throw std::system_error(errno, std::system_category(), "epoll_create1 failed");
        }

        // 创建用于唤醒事件循环的管道
        if (pipe(wake_pipe_) == -1) {
            close(epoll_fd_);
            throw std::system_error(errno, std::system_category(), "pipe creation failed");
        }

        // 监控管道的读端
        struct epoll_event ev;
        ev.events = EPOLLIN;
        ev.data.fd = wake_pipe_[0];
        if (epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, wake_pipe_[0], &ev) == -1) {
            close(wake_pipe_[0]);
            close(wake_pipe_[1]);
            close(epoll_fd_);
            throw std::system_error(errno, std::system_category(), "epoll_ctl failed");
        }
    }

    ~EPollReactor() override {
        stop();
        close(wake_pipe_[0]);
        close(wake_pipe_[1]);
        close(epoll_fd_);
    }

    void register_timer(std::function<void()> callback, 
                       std::chrono::milliseconds interval) override {
        std::lock_guard<std::mutex> lock(mutex_);
        timer_callback_ = std::move(callback);
        timer_interval_ = interval;
        next_timer_ = std::chrono::steady_clock::now() + interval;
    }

    void update_timer(std::chrono::milliseconds new_interval) override {
        std::lock_guard<std::mutex> lock(mutex_);
        timer_interval_ = new_interval;
        next_timer_ = std::chrono::steady_clock::now() + new_interval;
    }

    void unregister_timer() override {
        std::lock_guard<std::mutex> lock(mutex_);
        timer_callback_ = nullptr;
    }

    void register_fd(int fd, std::function<void(int)> callback) override {
        std::lock_guard<std::mutex> lock(mutex_);
        struct epoll_event ev;
        ev.events = EPOLLIN;  // 默认监听读事件
        ev.data.fd = fd;
        
        if (epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) == -1) {
            throw std::system_error(errno, std::system_category(), "epoll_ctl add failed");
        }
        
        fd_callbacks_[fd] = std::move(callback);
    }

    void unregister_fd(int fd) override {
        std::lock_guard<std::mutex> lock(mutex_);
        if (epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr) == -1) {
            throw std::system_error(errno, std::system_category(), "epoll_ctl del failed");
        }
        fd_callbacks_.erase(fd);
    }

    void set_fd_wake(int fd, bool wake_read, bool wake_write) override {
        std::lock_guard<std::mutex> lock(mutex_);
        struct epoll_event ev;
        ev.events = (wake_read ? static_cast<uint32_t>(EPOLLIN) : 0u) | 
                   (wake_write ? static_cast<uint32_t>(EPOLLOUT) : 0u);
        ev.data.fd = fd;
        
        if (epoll_ctl(epoll_fd_, EPOLL_CTL_MOD, fd, &ev) == -1) {
            throw std::system_error(errno, std::system_category(), "epoll_ctl mod failed");
        }
    }

    void register_async_callback(std::function<void()> callback) override {
        std::lock_guard<std::mutex> lock(mutex_);
        async_callbacks_.push(std::move(callback));
        // 写入一个字节到管道以唤醒事件循环
        char byte = 1;
        write(wake_pipe_[1], &byte, 1);
    }

    void notify_async_complete() override {
        // 在异步回调完成时调用
        char byte = 1;
        write(wake_pipe_[1], &byte, 1);
    }

    void acquire_mutex() override {
        mutex_.lock();
    }

    void release_mutex() override {
        mutex_.unlock();
    }

    void run() override {
        running_ = true;
        std::vector<epoll_event> events(64);

        while (running_) {
            auto now = std::chrono::steady_clock::now();
            std::chrono::milliseconds timeout(1000);  // 默认1秒超时

            {
                std::lock_guard<std::mutex> lock(mutex_);
                if (timer_callback_ && next_timer_ > now) {
                    timeout = std::chrono::duration_cast<std::chrono::milliseconds>(
                        next_timer_ - now);
                }
            }

            int nfds = epoll_wait(epoll_fd_, events.data(), events.size(), 
                                timeout.count());

            if (nfds == -1) {
                if (errno == EINTR) continue;
                throw std::system_error(errno, std::system_category(), "epoll_wait failed");
            }

            // 处理文件描述符事件
            for (int i = 0; i < nfds; ++i) {
                int fd = events[i].data.fd;
                
                if (fd == wake_pipe_[0]) {
                    // 处理异步回调
                    char buffer[256];
                    read(wake_pipe_[0], buffer, sizeof(buffer));
                    process_async_callbacks();
                } else {
                    // 处理其他文件描述符事件
                    std::lock_guard<std::mutex> lock(mutex_);
                    auto it = fd_callbacks_.find(fd);
                    if (it != fd_callbacks_.end()) {
                        it->second(fd);
                    }
                }
            }

            // 处理定时器
            now = std::chrono::steady_clock::now();
            std::lock_guard<std::mutex> lock(mutex_);
            if (timer_callback_ && now >= next_timer_) {
                timer_callback_();
                next_timer_ = now + timer_interval_;
            }
        }
    }

    void stop() override {
        running_ = false;
        notify_async_complete();  // 唤醒事件循环
    }

private:
    void process_async_callbacks() {
        std::queue<std::function<void()>> callbacks;
        {
            std::lock_guard<std::mutex> lock(mutex_);
            std::swap(callbacks, async_callbacks_);
        }

        while (!callbacks.empty()) {
            auto& callback = callbacks.front();
            callback();
            callbacks.pop();
        }
    }

private:
    bool running_;
    int epoll_fd_;
    int wake_pipe_[2];
    std::mutex mutex_;
    std::map<int, std::function<void(int)>> fd_callbacks_;
    std::queue<std::function<void()>> async_callbacks_;
    
    std::function<void()> timer_callback_;
    std::chrono::milliseconds timer_interval_;
    std::chrono::steady_clock::time_point next_timer_;
};

// 工厂函数实现
std::shared_ptr<Reactor> create_reactor() {
    return std::make_shared<EPollReactor>();
}

} // namespace klippy